package cn.codecang.mall.config;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jetbrains.annotations.NotNull;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.scheduling.annotation.Async;

import java.util.Optional;

/**
 * ConsumerConfiguration
 * Kafka 消费数据
 *
 * @author xiaoluo
 * @date 2023/6/6 18:38
 */
@Configuration
public class ConsumerConfiguration {
    @Async
    @KafkaListener(topicPartitions =
            {@TopicPartition(topic = "test",
                    partitions = {"0"})
            })
    public void consume(@NotNull ConsumerRecord<?,?> consumerRecord) {
        String topic = consumerRecord.topic();
        int partition = consumerRecord.partition();
        Object key = consumerRecord.key();
        Optional<Object> kafkaMassage = Optional.ofNullable(consumerRecord.value());
        if (kafkaMassage.isPresent()) {
            Object o = kafkaMassage.get();
            System.out.println("接收到的消息是：" + o + topic + partition + key);
        }
    }

    @Async
    @KafkaListener(topicPartitions =
            {@TopicPartition(topic = "topic.quick.init",
                    partitions = {"1"})
            })
    public void consume1(String message) {
        System.out.println("topic.quick.init 接收到消息 111：" + message);
    }

    @Async
    @KafkaListener(topicPartitions =
            {@TopicPartition(topic = "topic.quick.init",
                    partitions = {"2"})
            })
    public void consume2(String message) {
        System.out.println("topic.quick.init 接收到消息 222：" + message);
    }

    @Async
    @KafkaListener(topicPartitions =
            {@TopicPartition(topic = "topic.quick.init",
                    partitions = {"3"})
            })
    public void consume3(String message) {
        System.out.println("topic.quick.init 接收到消息 333：" + message);
    }
}
